public class RunRecursiveScheduling {
    /**
     * Generates an observable sequence by iterating a state from an initial 
     * state until the condition returns false.
     */
    public static <TState, R> OnSubscribeFunc<R> generate(
            final TState initialState,
            final Func1<TState, Boolean> condition,
            final Func1<TState, TState> iterate,
            final Func1<TState, R> resultSelector,
            final Scheduler scheduler) {
        return new OnSubscribeFunc<R>() {
            @Override
            public Subscription onSubscribe(final Observer<? super R> observer) {
                return scheduler.schedule(initialState, new Func2<Scheduler, TState, Subscription>() {
                    @Override
                    public Subscription call(Scheduler s, TState state) {
                        boolean hasNext;
                        try {
                            hasNext = condition.call(state);
                        } catch (Throwable t) {
                            observer.onError(t);
                            return Subscriptions.empty();
                        }
                        if (hasNext) {
                            R result;
                            try {
                                result = resultSelector.call(state);
                            } catch (Throwable t) {
                                observer.onError(t);
                                return Subscriptions.empty();
                            }
                            observer.onNext(result);

                            TState nextState;
                            try {
                                nextState = iterate.call(state);
                            } catch (Throwable t) {
                                observer.onError(t);
                                return Subscriptions.empty();
                            }

                            return s.schedule(nextState, this);
                        }
                        observer.onCompleted();
                        return Subscriptions.empty();
                    }
                });
            }
        };
    }

    public static void main(String[] args) throws Exception {
        Observable<Integer> source = Observable.create(generate(
                0, a -> true, a -> a + 1, a -> a, Schedulers.threadPoolForComputation()));

        CountDownLatch latch = new CountDownLatch(1);
        Subscription s = source.subscribe(v -> { 
                    if (v % 100_000 == 0) {
                        System.out.println(v);
                    }
                    if (v >= 10_000_000) { 
                        latch.countDown(); 
                    } 
                }, 
                e -> { e.printStackTrace(); latch.countDown(); }, 
                latch::countDown);

        latch.await();

        System.out.println("Wait done.");

        s.unsubscribe();

        System.out.println("Unsubscribe done.");
    }
}